/**
 * @ChannelEventHandler.java  COPYRIGHT xiaojian
 * 
 * Created on 2017年7月9日 下午7:44:25
 */
package xj.toolkit.access.connector;

import java.io.IOException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.ReferenceCountUtil;
import xj.toolkit.netty.transport.TransportUtils;
import xj.toolkit.netty.transport.endpoint.DefaultClientEndpoint;
import xj.toolkit.netty.transport.endpoint.Endpoint;
import xj.toolkit.access.msg.AccessSynMessage;
import xj.toolkit.ebus.EventBus;

/**
 * @author <a href="mailto:cxj2000@gmail.com">xiaojian.cao</a>
 *
 */
@ChannelHandler.Sharable
public abstract class AccessTcpConnectorChannelHandler extends ChannelInboundHandlerAdapter {

	private static final Logger logger = LoggerFactory.getLogger(AccessTcpConnectorChannelHandler.class);

	private String eventModuleMsgArrived;
	private String eventSessionOpened;
	private String eventSessionClosed;

	private EventBus eventBus;

	public String getEventModuleMsgArrived() {
		return eventModuleMsgArrived;
	}

	public void setEventModuleMsgArrived(String eventModuleMsgArrived) {
		this.eventModuleMsgArrived = eventModuleMsgArrived;
	}

	public String getEventSessionOpened() {
		return eventSessionOpened;
	}

	public void setEventSessionOpened(String eventSessionOpened) {
		this.eventSessionOpened = eventSessionOpened;
	}

	public String getEventSessionClosed() {
		return eventSessionClosed;
	}

	public void setEventSessionClosed(String eventSessionClosed) {
		this.eventSessionClosed = eventSessionClosed;
	}

	public EventBus getEventBus() {
		return eventBus;
	}

	public void setEventBus(EventBus eventBus) {
		this.eventBus = eventBus;
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		if (!(cause instanceof IOException)) {
			logger.error("exceptionCaught: ", cause);
			ctx.channel().close();
		}
	}

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		Endpoint endpoint = TransportUtils.getEndpointOfSession(ctx);
		if (null != endpoint) {
			endpoint.messageReceived(ctx, msg);
		} else {
			logger.warn("{} missing endpoint, ignore incoming msg:{}", msg);
		}
	

		if (null != eventModuleMsgArrived) {
			eventBus.fireEvent(eventModuleMsgArrived, msg);
		}

		// -- TODO(?)
		ReferenceCountUtil.release(msg);
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		Endpoint endpoint = TransportUtils.getEndpointOfSession(ctx);
		
		if (null != eventSessionClosed) {
			eventBus.fireEvent(eventSessionClosed, endpoint);
		}
		
		if (null != endpoint) {
			endpoint.stop();
		}
		
		ctx.fireChannelInactive();
	}

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		logger.info("channelActive:{}", ctx.channel());

		// create endpoint
		Endpoint endpoint = createEndpoint();
		if (null != endpoint) {
			TransportUtils.attachEndpointToSession(ctx, endpoint);
		}
		
		Channel channel = ctx.channel();
		if (null != endpoint && endpoint instanceof DefaultClientEndpoint) {
			((DefaultClientEndpoint)endpoint).setChannel(channel);
		}
		
		if (null != eventSessionOpened) {
			eventBus.fireEvent(eventSessionOpened, ctx.channel());
		}
	
		// -- 询问后端模块号以及实例号。
		AccessSynMessage message = new AccessSynMessage();
		channel.writeAndFlush(message);
	
	}

	protected abstract Endpoint createEndpoint();

}
